package com.atguigu.microblog;

import com.atguigu.microblog.bean.Message;
import com.atguigu.util.HBaseUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.CollectionUtils;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * @author lvbingbing
 * @date 2020-08-03 11:03
 */
@Slf4j
public class MicroBlogDemo {

    /**
     * 获取配置conf
     */
    private Configuration configuration;

    /**
     * 微博内容表的表名
     */
    private static final byte[] TABLE_CONTENT = Bytes.toBytes("weibo:content");

    /**
     * 用户关系表的表名
     */
    private static final byte[] TABLE_RELATIONS = Bytes.toBytes("weibo:relations");

    /**
     * 微博收件箱表的表名
     */
    private static final byte[] TABLE_RECEIVE_CONTENT_EMAIL = Bytes.toBytes("weibo:receive_content_email");

    @Before
    public void initConfiguration() {
        configuration = HBaseConfiguration.create();
        // zookeeper节点
        configuration.set("hbase.zookeeper.quorum", "hadoop102,hadoop103,hadoop104");
        // zookeeper端口
        configuration.set("hbase.zookeeper.property,clientPort", "2181");
    }

    /**
     * 1、创建命名空间以及表名的定义
     */
    @Test
    public void initNamespace() {
        try (
                // 创建连接
                Connection connection = ConnectionFactory.createConnection(configuration);
                // 管理员可用于创建，删除，列出，启用和禁用表，添加和删除表列族以及其他管理操作。
                Admin admin = connection.getAdmin();
        ) {
            //命名空间类似于关系型数据库中的schema，可以想象成文件夹
            NamespaceDescriptor.Builder builder = NamespaceDescriptor.create("weibo");
            builder.addConfiguration("creator", "Jinji").addConfiguration("create_time", System.currentTimeMillis() + "");
            NamespaceDescriptor namespaceDescriptor = builder.build();
            admin.createNamespace(namespaceDescriptor);

            HBaseUtil.checkCreateNamespace(admin, "weibo");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 2、创建微博内容表
     * <p>
     * Table Name：weibo:content
     * RowKey：用户ID_时间戳
     * ColumnFamily：info
     * ColumnLabel：标题	内容		图片URL
     * Version：1个版本
     */
    @Test
    public void createTableContent() {
        try (
                // 创建连接
                Connection connection = ConnectionFactory.createConnection(configuration);
                // 管理员可用于创建，删除，列出，启用和禁用表，添加和删除表列族以及其他管理操作。
                Admin admin = connection.getAdmin();
        ) {
            //创建表表述
            HTableDescriptor content = new HTableDescriptor(TableName.valueOf(TABLE_CONTENT));
            //创建列族描述
            HColumnDescriptor columnFamilyInfo = new HColumnDescriptor(Bytes.toBytes("info"));
            //设置块缓存
            columnFamilyInfo.setBlockCacheEnabled(true);
            //设置块缓存大小
            columnFamilyInfo.setBlocksize(2097152);
            //设置压缩方式
            columnFamilyInfo.setCompressionType(Compression.Algorithm.SNAPPY);
            //设置版本确界
            columnFamilyInfo.setMaxVersions(1);
            columnFamilyInfo.setMinVersions(1);

            content.addFamily(columnFamilyInfo);
            admin.createTable(content);

            HBaseUtil.validateTable(connection, admin, TableName.valueOf(TABLE_CONTENT));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 3、用户关系表
     * Table Name:weibo:relations0
     * RowKey:用户ID
     * ColumnFamily:attends,fans
     * ColumnLabel:关注用户ID，粉丝用户ID
     * ColumnValue:用户ID
     * Version：1个版本
     */
    @Test
    public void createTableRelations() {
        try (
                // 创建连接
                Connection connection = ConnectionFactory.createConnection(configuration);
                // 管理员可用于创建，删除，列出，启用和禁用表，添加和删除表列族以及其他管理操作。
                Admin admin = connection.getAdmin()
        ) {
            //关注的人的列族
            HColumnDescriptor columnFamilyAttends = new HColumnDescriptor(Bytes.toBytes("attends"));
            //设置块缓存
            columnFamilyAttends.setBlockCacheEnabled(true);
            //设置块缓存大小
            columnFamilyAttends.setBlocksize(2097152);
            //设置压缩方式
            columnFamilyAttends.setCompressionType(Compression.Algorithm.SNAPPY);
            //设置版本确界
            columnFamilyAttends.setMaxVersions(1);
            columnFamilyAttends.setMinVersions(1);

            //粉丝列族
            HColumnDescriptor columnFamilyFans = new HColumnDescriptor(Bytes.toBytes("fans"));
            columnFamilyFans.setBlockCacheEnabled(true);
            columnFamilyFans.setBlocksize(2097152);
            columnFamilyFans.setMaxVersions(1);
            columnFamilyFans.setMinVersions(1);

            HTableDescriptor relations = new HTableDescriptor(TableName.valueOf(TABLE_RELATIONS));
            relations.addFamily(columnFamilyAttends);
            relations.addFamily(columnFamilyFans);
            admin.createTable(relations);

            HBaseUtil.validateTable(connection, admin, TableName.valueOf(TABLE_RELATIONS));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 4、创建微博收件箱表
     * Table Name: weibo:receive_content_email
     * RowKey:用户ID
     * ColumnFamily:info
     * ColumnLabel:用户ID-发布微博的人的用户ID
     * ColumnValue:关注的人的微博的RowKey
     * Version:1000
     */
    @Test
    public void createTableReceiveContentEmail() {
        try (
                // 创建连接
                Connection connection = ConnectionFactory.createConnection(configuration);
                // 管理员可用于创建，删除，列出，启用和禁用表，添加和删除表列族以及其他管理操作。
                Admin admin = connection.getAdmin();
        ) {
            HColumnDescriptor columnFamilyInfo = new HColumnDescriptor(Bytes.toBytes("info"));
            columnFamilyInfo.setBlockCacheEnabled(true);
            columnFamilyInfo.setBlocksize(2097152);
            columnFamilyInfo.setMaxVersions(1000);
            columnFamilyInfo.setMinVersions(1000);

            HTableDescriptor receiveContentEmail = new HTableDescriptor(TableName.valueOf(TABLE_RECEIVE_CONTENT_EMAIL));
            receiveContentEmail.addFamily(columnFamilyInfo);
            admin.createTable(receiveContentEmail);

            HBaseUtil.validateTable(connection, admin, TableName.valueOf(TABLE_RECEIVE_CONTENT_EMAIL));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 5、发布微博内容
     */
    @Test
    public void testPublishContent() {
        publishContent("0001", "今天买了一包空气，送了点薯片，非常开心！！");
        publishContent("0001", "今天天气不错。");
    }

    /**
     * 6、添加关注
     */
    @Test
    public void testAddAttend() {
        publishContent("0008", "准备下课！");
        publishContent("0009", "准备关机！");
        addAttends("0001", "0008", "0009");
    }

    /**
     * 7、取消关注
     */
    @Test
    public void testRemoveAttend() {
        removeAttends("0001", "0008");
    }

    /**
     * 8、展示内容
     */
    @Test
    public void testShowMessage() {
        List<Message> messages = getAttendsContent("0001");
        for (Message message : messages) {
            log.info(message.toString());
        }
    }

    /**
     * 发布微博
     * a、微博内容表中数据+1
     * b、向微博收件箱表中加入微博的RowKey
     */
    public void publishContent(String uid, String content) {
        try (Connection connection = ConnectionFactory.createConnection(configuration)) {
            //a、微博内容表中添加1条数据，首先获取微博内容表描述
            Table contentTable = connection.getTable(TableName.valueOf(TABLE_CONTENT));
            //组装RowKey
            long timestamp = System.currentTimeMillis();
            String rowKey = uid + "_" + timestamp;
            Put put = new Put(Bytes.toBytes(rowKey));
            put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("content"), timestamp, Bytes.toBytes(content));
            contentTable.put(put);

            //b、向微博收件箱表中加入发布的RowKey
            //b.1、查询用户关系表，得到当前用户有哪些粉丝
            Table relationsTable = connection.getTable(TableName.valueOf(TABLE_RELATIONS));
            //b.2、取出目标数据
            Get get = new Get(Bytes.toBytes(uid));
            get.addFamily(Bytes.toBytes("fans"));

            Result result = relationsTable.get(get);
            List<byte[]> fans = new ArrayList<>();

            //遍历取出当前发布微博的用户的所有粉丝数据
            for (Cell cell : result.rawCells()) {
                fans.add(CellUtil.cloneQualifier(cell));
            }
            //如果该用户没有粉丝，则直接return
            if (CollectionUtils.isEmpty(fans)) {
                return;
            }
            //开始操作收件箱表
            Table receiveContentEmail = connection.getTable(TableName.valueOf(TABLE_RECEIVE_CONTENT_EMAIL));
            List<Put> puts = new ArrayList<>();
            for (byte[] fan : fans) {
                Put fanPut = new Put(fan);
                fanPut.addColumn(Bytes.toBytes("info"), Bytes.toBytes(uid), timestamp, Bytes.toBytes(rowKey));
                puts.add(fanPut);
            }
            receiveContentEmail.put(puts);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 关注用户逻辑
     * a、在微博用户关系表中，对当前主动操作的用户添加新的关注的好友
     * b、在微博用户关系表中，对被关注的用户添加粉丝（当前操作的用户）
     * c、当前操作用户的微博收件箱添加所关注的用户发布的微博rowKey
     */
    public void addAttends(String uid, String... attends) {
        //参数过滤
        if (attends == null || attends.length <= 0 || uid == null || uid.length() <= 0) {
            return;
        }
        try (Connection connection = ConnectionFactory.createConnection(configuration)) {

            //用户关系表操作对象（连接到用户关系表）
            Table relationsTable = connection.getTable(TableName.valueOf(TABLE_RELATIONS));
            List<Put> puts = new ArrayList<>();
            //a、在微博用户关系表中，添加新关注的好友
            Put attendPut = new Put(Bytes.toBytes(uid));
            for (String attend : attends) {
                //为当前用户添加关注的人
                attendPut.addColumn(Bytes.toBytes("attends"), Bytes.toBytes(attend), Bytes.toBytes(attend));
                //b、为被关注的人，添加粉丝
                Put fansPut = new Put(Bytes.toBytes(attend));
                fansPut.addColumn(Bytes.toBytes("fans"), Bytes.toBytes(uid), Bytes.toBytes(uid));
                //将所有关注的人一个一个的添加到puts（List）集合中
                puts.add(fansPut);
            }
            puts.add(attendPut);
            relationsTable.put(puts);

            //c.1、微博收件箱添加关注的用户发布的微博内容（content）的rowKey
            Table contentTable = connection.getTable(TableName.valueOf(TABLE_CONTENT));
            Scan scan = new Scan();
            //用于存放取出来的关注的人所发布的微博的rowKey
            List<byte[]> rowKeys = new ArrayList<>();

            for (String attend : attends) {
                //过滤扫描rowKey，即：前置位匹配被关注的人的uid_
                RowFilter filter = new RowFilter(CompareFilter.CompareOp.EQUAL, new SubstringComparator(attend + "_"));
                //为扫描对象指定过滤规则
                scan.setFilter(filter);
                //通过扫描对象得到scanner
                ResultScanner result = contentTable.getScanner(scan);
                //迭代器遍历扫描出来的结果集
                for (Result r : result) {
                    //取出每一个符合扫描结果的那一行数据
                    for (Cell cell : r.rawCells()) {
                        //将得到的rowKey放置于集合容器中
                        rowKeys.add(CellUtil.cloneRow(cell));
                    }
                }
            }

            //c.2、将取出的微博rowKey放置于当前操作用户的收件箱中
            if (CollectionUtils.isEmpty(rowKeys)) {
                return;
            }
            //得到微博收件箱表的操作对象
            Table receiveContentEmail = connection.getTable(TableName.valueOf(TABLE_RECEIVE_CONTENT_EMAIL));
            //用于存放多个关注的用户的发布的多条微博rowKey信息
            List<Put> recPuts = new ArrayList<>();
            for (byte[] rk : rowKeys) {
                Put put = new Put(Bytes.toBytes(uid));
                //uid_timestamp
                String rowKey = Bytes.toString(rk);
                //截取uid
                String attendUid = rowKey.substring(0, rowKey.indexOf("_"));
                long timestamp = Long.parseLong(rowKey.substring(rowKey.indexOf("_") + 1));
                //将微博rowKey添加到指定单元格中
                put.addColumn(Bytes.toBytes("info"), Bytes.toBytes(attendUid), timestamp, rk);
                recPuts.add(put);
            }

            receiveContentEmail.put(recPuts);

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 取消关注（remove)
     * a、在微博用户关系表中，对当前主动操作的用户删除对应取关的好友
     * b、在微博用户关系表中，对被取消关注的人删除粉丝（当前操作人）
     * c、从收件箱中，删除取关的人的微博的rowKey
     */
    public void removeAttends(String uid, String... attends) {
        //过滤数据
        if (uid == null || uid.length() <= 0 || attends == null || attends.length <= 0) {
            return;
        }

        try (Connection connection = ConnectionFactory.createConnection(configuration)) {

            //a、在微博用户关系表中，删除已关注的好友
            Table relations = connection.getTable(TableName.valueOf(TABLE_RELATIONS));

            //待删除的用户关系表中的所有数据
            List<Delete> deletes = new ArrayList<>();
            //当前取关操作者的uid对应的Delete对象
            Delete attendDelete = new Delete(Bytes.toBytes(uid));
            //遍历取关，同时每次取关都要将被取关的人的粉丝-1
            for (String attend : attends) {
                attendDelete.addColumn(Bytes.toBytes("attends"), Bytes.toBytes(attend));
                //b
                Delete fansDelete = new Delete(Bytes.toBytes(attend));
                fansDelete.addColumn(Bytes.toBytes("fans"), Bytes.toBytes(uid));
                deletes.add(fansDelete);
            }

            deletes.add(attendDelete);
            relations.delete(deletes);

            //c、删除取关的人的微博rowKey从收件箱表中
            Table receiveContentEmail = connection.getTable(TableName.valueOf(TABLE_RECEIVE_CONTENT_EMAIL));

            Delete recDelete = new Delete(Bytes.toBytes(uid));
            for (String attend : attends) {
                recDelete.addColumn(Bytes.toBytes("info"), Bytes.toBytes(attend));
            }
            receiveContentEmail.delete(recDelete);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 获取微博实际内容
     * a、从微博收件箱中获取所有关注的人的发布的微博的rowKey
     * b、根据得到的rowKey去微博内容表中得到数据
     * c、将得到的数据封装到Message对象中
     */
    public List<Message> getAttendsContent(String uid) {
        try (Connection connection = ConnectionFactory.createConnection(configuration)) {
            Table receiveContentEmail = connection.getTable(TableName.valueOf(TABLE_RECEIVE_CONTENT_EMAIL));
            //a、从收件箱中取得微博rowKey
            Get get = new Get(Bytes.toBytes(uid));
            //设置最大版本号
            get.setMaxVersions(5);
            List<byte[]> rowKeys = new ArrayList<>();
            Result result = receiveContentEmail.get(get);
            for (Cell cell : result.rawCells()) {
                rowKeys.add(CellUtil.cloneValue(cell));
            }
            //b、根据取出的所有rowKey去微博内容表中检索数据
            Table contentTable = connection.getTable(TableName.valueOf(TABLE_CONTENT));
            List<Get> gets = new ArrayList<>();
            //根据rowKey取出对应微博的具体内容
            for (byte[] rk : rowKeys) {
                Get g = new Get(rk);
                gets.add(g);
            }
            //得到所有的微博内容的result对象
            Result[] results = contentTable.get(gets);

            List<Message> messages = new ArrayList<>();
            for (Result res : results) {
                for (Cell cell : res.rawCells()) {
                    Message message = new Message();
                    String rowKey = Bytes.toString(CellUtil.cloneRow(cell));
                    String userid = rowKey.substring(0, rowKey.indexOf("_"));
                    String timestamp = rowKey.substring(rowKey.indexOf("_") + 1);
                    String content = Bytes.toString(CellUtil.cloneValue(cell));

                    message.setContent(content);
                    message.setTimestamp(timestamp);
                    message.setUid(userid);

                    messages.add(message);
                }
            }
            return messages;
        } catch (IOException e) {
            e.printStackTrace();
            return new ArrayList<>();
        }
    }
}
